package day11

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 3.3.4	Kafka 0-10 Direct模式
 *
 * 1）需求：通过SparkStreaming从Kafka读取数据，并将读取过来的数据做简单计算，最终打印到控制台。
 * 2）导入依赖，为了避免和0-8冲突，我们新建一个module演示
 *
 * 消费的offset保存在 __consumer_offsets 主题中
 *
 * 3）通过以下命令查看
 * $KAFKA_HOME/bin/kafka-topics.sh --list --bootstrap-server hadoop102:9092
 *
 */
object Spark_Streaming_3_3_4 {
  def main(args: Array[String]): Unit = {
    // 1、创建SparkConf对象
    val conf: SparkConf = new SparkConf().setAppName("Spark_Streaming_3_3_3").setMaster("local[*]")
    // 2、创建SparkStreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 3、消费kafka数据创建流
    val kafkaParams: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "bigdata",
      // ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer]
    )
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc = ssc,
      // 在大多数情况下，它将一致地在所有执行器上分布分区。
      locationStrategy = LocationStrategies.PreferConsistent,
      consumerStrategy = ConsumerStrategies.Subscribe[String, String](topics = Set("my-bak"), kafkaParams)
    )
    // 4、计算wordCount并打印
    kafkaDStream.map(consumerRecord => consumerRecord.value())
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      .print()
    // 5、启动任务
    ssc.start()
    ssc.awaitTermination()
  }
}

/**
 * 3.3.5	消费Kafka数据模式总结
 *
 * 	0-8 ReceiverAPI:
 * ①专门的Executor读取数据，速度不统一
 * ②跨机器传输数据，WAL
 * ③Executor读取数据通过多个线程的方式，想要增加并行度，则需要多个流unoin
 * ④offset存储在zookeeper中
 *
 * 	0-8 DirectAPI:
 * ①Executor读取数据并计算
 * ②增加Executor个数来增加消费的并行度
 * ③offset存储：
 * a、checkpoint(getActiveOrCreate方式创建StreamingContext)
 * b、手动维护(有事务的存储系统)
 * c、获取offset必须在第一个调用的算子中：offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
 *
 * 	0-10 DirectAPI:
 * ①
 * ②Executor读取并计算
 * ③增加Executor个数来增加消费的并行度
 * ④offset存储
 * a、_consumer_offsets系统主题中
 * b、手动维护(有事务的存储系统)
 */
object Spark_Streaming_3_3_5 {}
